Flink: Close catalog opened by the dynamic sink serializer cache - #17437
Flink: Close catalog opened by the dynamic sink serializer cache#17437vishnuprakaz wants to merge 7 commits into
Conversation
mxm
left a comment
mxm
left a comment
There was a problem hiding this comment.
Thanks for the PR @vishnuprakaz!
| Catalog catalog = mock(Catalog.class, withSettings().extraInterfaces(Closeable.class)); | ||
| when(catalog.loadTable(any(TableIdentifier.class))).thenReturn(table); | ||
| CatalogLoader catalogLoader = mock(CatalogLoader.class); | ||
| when(catalogLoader.loadCatalog()).thenReturn(catalog); | ||
| cache = new TableSerializerCache(catalogLoader, 10); |
There was a problem hiding this comment.
Could we write this test without using Mockito?
| } | ||
| } | ||
|
|
||
| private static class SingleCatalogLoader implements CatalogLoader { |
There was a problem hiding this comment.
Can we replace this with HadoopCatalogExtension?
| if (catalog instanceof Closeable) { | ||
| try { | ||
| ((Closeable) catalog).close(); | ||
| } catch (IOException e) { | ||
| LOG.warn("Failed to close catalog {}", catalog.name(), e); | ||
| } | ||
| } |
There was a problem hiding this comment.
Would it be feasible to let the CatalogLoader manage the lifecycle of the catalog and reuse it when necessary? It is a bit odd to check for Closable when the Catalog interface doesn't contain this interface.
There was a problem hiding this comment.
On this I looked into having the loader manage and reuse the catalog. the problem I ran into is that the serializer has no teardown hook, so as far as I can tell nothing at this layer could ever close a held catalog. Does that match your thinking that the reuse part belongs in the TaskManager level cache?
While checking, I noticed loadCatalog() has six other call sites in the sink that don't cloae the catalog either. The operator-level ones do have close() hooks, would it be worth a similar interim fix for those, or better to leave them for the TaskManager level cache work?
For the Closeable check here.... would it make sense to switch update() to TableLoader.fromCatalog() in a try with resources, like FlinkSink and IcebergSink do for one shot loads?
but the lifecycle (and the Closeable check) would live inside TableLoader instead of this class.
| try (TableLoader tableLoader = | ||
| TableLoader.fromCatalog(catalogLoader, TableIdentifier.parse(tableName))) { | ||
| tableLoader.open(); | ||
| Table table = tableLoader.loadTable(); | ||
| schemas = table.schemas(); | ||
| specs = table.specs(); | ||
| } catch (IOException e) { | ||
| // only close() throws IOException here; a failed close should not fail the lookup | ||
| LOG.warn("Failed to close catalog for table {}", tableName, e); | ||
| } |
There was a problem hiding this comment.
I'm not sure loading / closing on every update is a good idea. Can we store Catalog in a field? This will get rid of the duplicate loading and only load once per job.
If we are concerned about not closing one instance, we could close it via RuntimeContext#registerUserCodeClassLoaderReleaseHookIfAbsent, but I'm not sure this is necessary.
There was a problem hiding this comment.
That makes a lot of sense, loading once and reusing is better trade actually. I was thinking per table and missed how many tables the dynamic sink could be routing. Made the changes, happy to adjust if anything looks off :)
2 participants
The dynamic sink's
TableSerializerCacheloads a catalog withCatalogLoader.loadCatalog()on every cache miss (an unknown schema or spec id, or an LRU eviction) but never closes it. Each call builds a fresh catalog whose resources, such as a REST HTTP client and thread pool or a Hive metastore client pool, are released only on close, so a long-running job that keeps missing the cache leaks connections and threads.The cache lives inside a Flink
TypeSerializerthat has no teardown hook, so it now closes the catalog right after reading the table's schemas and specs. This matches howTableLoadercloses the catalogs it opens.